package com.xuanyue.db.xuan.core.table;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.xuanyue.db.xuan.core.exception.IndexException;
import com.xuanyue.db.xuan.core.exception.SQLException;
import com.xuanyue.db.xuan.core.index.BitIndex;
import com.xuanyue.db.xuan.core.index.ULongIndex;
import com.xuanyue.db.xuan.core.table.expr.Expr;
import com.xuanyue.db.xuan.core.table.expr.ExprValue;
import com.xuanyue.db.xuan.core.table.expr.ExprValueType;
import com.xuanyue.db.xuan.core.table.expr.SortTemplate;
import com.xuanyue.db.xuan.core.tools.SafeManager;
import com.xuanyue.db.xuan.core.tools.Savor;
import com.xuanyue.db.xuan.msg.VLAUETYPE;

/**
 * 主表数据
 * 
 * 警告：
 * 写入暂时不提供并发性，当前是串行写入
 * @author guanh
 */
public class XyTable implements IXyTable{
	private static Logger log = LoggerFactory.getLogger(XyTable.class);
	
	
	/**表名*/
	private String name;
	/**持久化路径*/
	private String path;
	/**数据列*/
	protected Map<String,IColumn> name2column = new HashMap<>();
	/**最大 id*/
	private AtomicInteger maxId = new AtomicInteger();
	/**内存管理器   ，银行家算法实现*/
	private SafeManager<IBitIndex> manager;
	/**记录是否存在*/
	protected IBitIndex mask = new BitIndex();
	/** id保护区 <br/>
	 * 进入保护区的数据，不可以被修改和读取，但是仍然可以参与其他sql的条件过滤
	 */
	private Map<Integer,IdLock> idReserve = new HashMap<>();
	//private Map<Integer,Long> idWriteReserve = new HashMap<>();
	/**id保护区 锁*/
	private ReentrantLock lock = new ReentrantLock();
	private static final int SPLIT_CONDITION_NUM = 1000;
	private Map<Integer,Condition> split2Condition = new HashMap<>();
//	private Condition condition = lock.newCondition();
	/**数据版本*/
	private AtomicLong dataVersion = new AtomicLong();
	
	private void initDataVersioin() {
		ULongIndex ver = this.getColumn(VERSION_COLUMN);
		dataVersion.set(ver.max());
	}
	
	public XyTable(String name) {
		this(80,name,null);
	}
	public XyTable(String name,String path) {
		this(80,name,path);
	}
	public XyTable(int querySource,String name,String path) {
		this.name = name;
		this.path = path;
		BitIndexFactory f = new BitIndexFactory();
		f.setMaxId(maxId);
		manager = new SafeManager<IBitIndex>(f,querySource);

		for(int i=0;i<SPLIT_CONDITION_NUM;i++) {
			split2Condition.put(i, lock.newCondition());
		}
		mask.setPath(String.format("%s/mask",path));
	}
	
	@Override
	public void setPath(String path) {
		this.path = path;
		name2column.forEach( (k,v)->{
			v.setPath(String.format("%s/%s", path,k));
		});
	}
	
	
	@Override
	public IBitIndex getMask() {
		return mask;
	}

	@Override
	public long getCurrentVersion() {
		return dataVersion.get();
	}
	
	@Override
	public String getName() {
		return this.name;
	}
	@Override
	public Map<String,IColumn> getName2column(){
		return name2column;
	}
	@Override
	public void addColumn(String name,IColumn column) {
		column.setPath(String.format("%s/%s", path,name.toLowerCase()));
		name2column.put(name.toLowerCase(), column);
	}
	@Override
	public void dropColumn(String cname) {
		File data = new File(String.format("%s/%s", path,name.toLowerCase()));
		if(data.exists()) {
			data.delete();
		}
		name2column.remove( cname.toLowerCase());
	}
	
	@Override
	public List<IBitIndex> apply(int maxSouce) {
		return manager.apply(maxSouce);
	}
	@Override
	public void returnSource(List<IBitIndex> sources){
		manager.returnSource(sources);
	}

	@Override
	public void expr(Expr expr,List<IBitIndex> caches) {
		String cname = expr.getCname();
		name2column.get(cname).exe(expr, caches);
	}
	
	@Override
	public int checkAndContSource(Expr expr) {
		IColumn c = this.getColumn(expr.getCname());
		if(c==null) {
			return 0;
		}else {
			return c.checkAndContSource(expr);
		}
	}
	/**
	 * id 进入保护区
	 * @param id
	 * @return
	 * @author 解观海
	 * @throws InterruptedException 
	 * @date 2021年1月1日
	 */
	@Override
	public void id2Reserve(String key,int id,boolean write) {
		try {
			lock.lock();
			if(write) {
				if(idReserve.containsKey(id/64)) {
					while(idReserve.containsKey(id/64)) {
						try {
							split2Condition.get((id/64)% SPLIT_CONDITION_NUM ).await();
						} catch (InterruptedException e) {
							throw new SQLException(e);
						}
					}
				}
				idReserve.put(id/64, new IdLock(true));
			}else {
				while(true) {
					IdLock il = idReserve.get(id/64);
					if(il!=null) {
						if(il.isWriteLock()){
							try {
								split2Condition.get((id/64)% SPLIT_CONDITION_NUM ).await();
							} catch (InterruptedException e) {
								throw new SQLException(e);
							}
							continue;
						}else {
							il.putTicket(key);
						}
					}else {
						il = new IdLock(false);
						il.putTicket(key);
						idReserve.put(id/64, il);
					}
					break;
				}
			}
		} finally {
			lock.unlock();
		}
	}
	
	/**
	 * 1、优先获得数据空位（表中间的数据被删除后形成空位）<br/>
	 * 2、无空位时扩展一个新位置。
	 * @return
	 * @author 解观海
	 * @date 2021年1月1日
	 */
	private int nextRowId() {
		/**
		 * 1、优先获得数据空位（表中间的数据被删除后形成空位）
		 */
		try {
			lock.lock();
			int index=0;
			for(int i=0,s=idReserve.size();i<s;i++) {
				index = mask.nextClearBit(index);//找到一个空白位置
				if(-1!=index) {//有空白位置
					if(!idReserve.containsKey(index/64)) {//没有进入保护区
						idReserve.put(index/64,new IdLock(true));//写保护
						return index;
					}
				}else {
					//没有空白
					break;
				}
			}
			/**
			 * 无空位时扩展一个位置。
			 */
			index = maxId.getAndAdd(1);
			idReserve.put(index/64,new IdLock(true));
			return index;
		} finally {
			lock.unlock();
		}
		
	}
	
	private List<Integer> nextRowIdBatch(int num) {
		List<Integer> r = new ArrayList<>(num);
		/**
		 * 1、优先获得数据空位（表中间的数据被删除后形成空位）
		 */
		try {
			lock.lock();
			int index=0;
			int n=0;
			out:for(;n<num;n++) {
				for(int i=0,s=idReserve.size();i<s;i++) {
					index = mask.nextClearBit(index);//找到一个空白位置
					if(-1!=index) {//有空白位置
						if(!idReserve.containsKey(index/64)) {//没有进入保护区
							idReserve.put(index/64,new IdLock(true));//写保护
							r.add(index);
							continue out;
						}
					}else {
						//没有空白
						break out;
					}
				}
				break out;
			}
			for(;n<num;n++) {
				/**
				 * 无空位时扩展一个位置。
				 */
				index = maxId.getAndAdd(1);
				idReserve.put(index/64,new IdLock(true));
				r.add(index);
			}
			return r;
		} finally {
			lock.unlock();
		}
		
	}
	
	public void freeIdBatch(List<Integer> indexs) {
		try {
			lock.lock();
			for(int index:indexs) {
				idReserve.remove(index/64);
				split2Condition.get((index/64)% SPLIT_CONDITION_NUM ).signalAll();
			}
		} finally {
			lock.unlock();
		}
	}
	
	/**
	 * id 移出保护区
	 * @param index
	 * @author 解观海
	 * @date 2021年1月1日
	 */
	@Override
	public void freeId(String key,int index,boolean write) {
		try {
			lock.lock();
			if(write) {
				idReserve.remove(index/64);
				split2Condition.get((index/64)% SPLIT_CONDITION_NUM ).signalAll();
			}else {
				IdLock rl = idReserve.get(index/64);
				if(rl!=null) {
					rl.mvTicket(key);
					if(rl.isEmpty()) {
						idReserve.remove( index/64);
						split2Condition.get((index/64)% SPLIT_CONDITION_NUM ).signalAll();
					}
				}else {
					log.warn(String.format("id %s is out of reserve ", index));
				}
			}
			
		} finally {
			lock.unlock();
		}
	}
	@Override
	public int insertInto(Map<String,ExprValue> vs) {
		if(vs==null) {
			throw new SQLException("insertInto value is null");
		}
		String key = UUID.randomUUID().toString();
		int rowId = nextRowId();
		try {
			//1、数据版本
			ExprValue ev = new ExprValue();
			ev.setType(ExprValueType.LONG);
			ev.setValue(dataVersion.addAndGet(1));
			vs.put(VERSION_COLUMN,ev );
			//2、数据插入
			for(Entry<String, IColumn> en:name2column.entrySet()) {
				if(log.isDebugEnabled())
					log.debug(String.format("to insert into column %s", en.getKey()));
				en.getValue().set(rowId, 
						vs.containsKey(en.getKey())?vs.get(en.getKey()):ExprValue.getNULLExprValue()
						);
				en.getValue().flush(rowId);
			}
			mask.set(rowId);
			mask.flush(rowId);
		} catch (Exception e) {
			log.error("ERROR:",e);
			throw new SQLException(e.getMessage());
		}finally {
			freeId(key,rowId,true);
		}
		
		return rowId;
	}
	
	public List<ExprValue> insertIntoBatch(List<Map<String,ExprValue>> vsl) {
		if(vsl==null) {
			throw new SQLException("insertInto value is null");
		}//
		List<ExprValue> r = new ArrayList<>();
		List<Integer> rowIds = nextRowIdBatch(vsl.size());
		try {
			//1、数据版本
			ExprValue ev = new ExprValue();
			ev.setType(ExprValueType.LONG);
			int i=0;
			int rowId;
			for(Map<String,ExprValue> vs:vsl) {
				ev.setValue(dataVersion.addAndGet(1));
				vs.put(VERSION_COLUMN,ev );
				rowId = rowIds.get(i++);
				//2、数据插入
				for(Entry<String, IColumn> en:name2column.entrySet()) {
					if(log.isDebugEnabled())
						log.debug(String.format("to insert into column %s", en.getKey()));
					en.getValue().set(rowId, 
							vs.containsKey(en.getKey())?vs.get(en.getKey()):ExprValue.getNULLExprValue()
							);
					en.getValue().flush(rowId);
				}
				mask.set(rowId);
				mask.flush(rowId);
				r.add(new ExprValue(ExprValueType.INT,rowId));
			}
			return r;
		} catch (Exception e) {
			log.error("ERROR:",e);
			throw new SQLException(e.getMessage());
		}finally {
			freeIdBatch(rowIds);
		}
	}
	
	
	@Override
	public String insertIntoCheck(Map<String,ExprValue> vs) {
		if(vs==null)
			return "[record is null]";
		for(Entry<String,IColumn> e:name2column.entrySet()) {
			if(vs.containsKey(e.getKey() )) {
				if(!e.getValue().checkType(  vs.get(e.getKey()) )){
					return e.getKey();
				}
			}
		}
		return null;
	}
	@Override
	public int updateByRowid(Map<String,ExprValue> vs,int rowId,long version) {
		if(vs==null) {
			throw new IndexException("the value to set is null");
		}
		String key = UUID.randomUUID().toString();
		try {
			//lock.lock();
			id2Reserve(key, rowId, true);
			
			ULongIndex vc = this.getColumn(VERSION_COLUMN);
			if(vc.get(rowId)<=version) {
				//1、数据版本
				ExprValue ev = new ExprValue();
				ev.setType(ExprValueType.LONG);
				ev.setValue(dataVersion.addAndGet(1));
				vs.put(VERSION_COLUMN,ev );
				//2、数据插入
				name2column.forEach( (k,v)->{
					if(vs.containsKey(k)) {
						v.set(rowId, vs.get(k));
						v.flush(rowId);
					}
				});
				mask.set(rowId);
				mask.flush(rowId);
				this.saveRow(rowId);
				return 1;
			}
			return 0;
		} finally {
			freeId(key, rowId, true);
		}
	}
	@Override
	public int deleteByRowid(int rowid,long version) {
		String key = UUID.randomUUID().toString();
		try {
			id2Reserve(key, rowid, true);
			ULongIndex vc = this.getColumn(VERSION_COLUMN);
			if(mask.get(rowid)&&vc.get(rowid)<=version) {
				mask.set(rowid,false);
				ExprValue valueO = new ExprValue();
				valueO.setType(ExprValueType.LONG);
				valueO.setValue(dataVersion.addAndGet(1));
				vc.set(rowid, valueO);
				return 1;
			}else {
				return 0;
			}
		} finally {
			freeId(key, rowid, true);
		}
	}
	
	
	@Override
	public void flush(int rowId) {
		name2column.forEach( (k,v)->{
			v.flush(rowId);
		} );
		mask.flush(rowId);
	}
	
	private boolean checkVersion(int rowId,long versionNow) {
		ULongIndex vc = this.getColumn(VERSION_COLUMN);
		return vc.get(rowId)<=versionNow;
	}
	
	private Map<String,ExprValue> read(int rowId,Set<String> fields,List<ISortElement> sortElementList){
		if(!mask.get(rowId) ) {
			return null;
		}
		
		Map<String,ExprValue> r = new HashMap<>();
//		name2column.forEach( (k,v)->{
//			if(fields.contains(k))
//				r.put(k, v.getExprValue(rowId));
//		} );
		
		for(String f:fields) {
			if(name2column.containsKey(f.toLowerCase())) {
				r.put(f.toLowerCase(), name2column.get(f.toLowerCase()).getExprValue(rowId));
			}
		}
		
		r.put("rowid", new ExprValue(ExprValueType.INT, rowId));
		
		if(sortElementList!=null) {
			ISortElement se = null;
			byte[] sort = new byte[sortElementList.size()];
			byte t = 1;
			byte f = 0;
			for(int i=0,s=sortElementList.size();i<s;i++) {
				se = sortElementList.get(i);
				if(se.isDesc()) {
					sort[i] = (se.getData().get(rowId)?t:f);
				}else {
					sort[i] = (se.getData().get(rowId)?f:t);
				}
			}
			r.put(SORT_OBJECT_NAME, new ExprValue(ExprValueType.BYTEARRAY, sort));
		}
		return r;
	}
	
	@Override
	public Map<String,ExprValue> readByIdReserve(int rowId,Set<String> fields,List<ISortElement> sortElementList,long versionNow){
		String key = UUID.randomUUID().toString();
		try {
			id2Reserve(key, rowId, false);
			if(!checkVersion(rowId, versionNow )) {
				return null;
			}
			return read(rowId, fields, sortElementList);
		} finally {
			freeId(key, rowId, false);
		}
	}
	
	@Override
	public int getDataSize() {
		int r =0;
		for(Entry<String,IColumn> en:name2column.entrySet()) {
			r+=en.getValue().getDataSize();
		}
		return r;
	}
	
	public int getMaxId() {
		return maxId.get();
	}
	@Override
	public void save() {
		for(Entry<String,IColumn> en:name2column.entrySet()) {
			en.getValue().save();
		}
		Savor.write(maxId, String.format("%s/maxId",path));
		mask.save();
	}
	@Override
	public void saveRow(int rowId){
		for(Entry<String,IColumn> en:name2column.entrySet()) {
			en.getValue().saveRow(rowId);
		}
		Savor.write(maxId, String.format("%s/maxId",path));
		mask.saveRow(rowId);
	}
	@Override
	public void load()  {
		File p = new File(path);
		if(!p.exists()) {
			p.mkdirs();
			return;
		}
		for(Entry<String,IColumn> en:name2column.entrySet()) {
			log.debug(String.format("to load: %s/%s", path,en.getKey()));
			en.getValue().load();
		} 
		AtomicInteger maxId=Savor.read( String.format("%s/maxId",path));
		for(Entry<String,IColumn> en:name2column.entrySet()) {
			en.getValue().flush(maxId.get());
		}
		try {
			mask.load();
		} catch (Exception e) {
			throw new SQLException(e);
		}
		this.maxId.set(maxId.get());
		initDataVersioin();
	}
	@Override
	public void init()  {
		File p = new File(path);
		if(!p.exists()) {
			p.mkdirs();
		}
		File init_lock = new File(String.format("%s/init_lock", path));
		if(init_lock.exists()) {
			return;
		}
		for(Entry<String,IColumn> en:name2column.entrySet()) {
			p = new File(String.format("%s/%s", path,en.getKey()));
			p.mkdirs();
			en.getValue().init();
		}
		mask.init();
		Savor.write(maxId, String.format("%s/maxId",path));
		
		try {
			init_lock.createNewFile();
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}
	@Override
	public VLAUETYPE getType(String fn) {
		IColumn column = name2column.get(fn.toLowerCase());
		return column.getType();
	}
	@Override
	public List<ISortElement> getSortE(SortTemplate sortTemplate){
		IColumn column = name2column.get(sortTemplate.getCname());
		return column.getSortE(sortTemplate);
	}
	
	@SuppressWarnings("unchecked")
	@Override
	public <T extends IColumn>T getColumn(String cname) {
		return cname==null?null:(T)name2column.get(cname.toLowerCase());
	}
	@Override
	public void setSource(int source) {
		manager.setSource(source);
	}
	
}
class IdLock{
	private long create = System.currentTimeMillis();
	private Map<String,Long> tickets;// = new HashMap<>();
	private boolean isWrite;
	
	public IdLock(boolean isWrite) {
		this.isWrite = isWrite;
	}
	
	public long getCreate() {
		return create;
	}
	public void setCreate(long create) {
		this.create = create;
	}
	public Map<String, Long> getTickets() {
		return tickets;
	}
	public void setTickets(Map<String, Long> tickets) {
		this.tickets = tickets;
	}
	
	public void putTicket(String key) {
		if(tickets==null)
			tickets = new HashMap<>();
		tickets.put(key, System.currentTimeMillis());
	}
	public void mvTicket(String key) {
		if(tickets!=null) {
			tickets.remove(key);
		}
	}
	public boolean isEmpty() {
		return tickets==null||tickets.isEmpty();
	}
	public boolean isWriteLock() {
		return isWrite;
	}
}
